package com.smsc.headend.connector.service.impl;

import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.bo.ConnectionInfo;
import com.smsc.headend.connector.builder.LoggerBuilder;
import com.smsc.headend.connector.mapper.HdComTrafficMapper;
import com.smsc.headend.connector.mapper.LogDevicePacketMapper;
import com.smsc.headend.connector.mapper.LogTerminalAccessMapper;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.service.LogTerminalAccessService;
import com.smsc.headend.connector.service.TerminalChannelService;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.data.dto.TaskPacketDTO;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.report.entity.LogDevicePacket;
import com.smsc.headend.module.request.entity.HdComTraffic;
import com.smsc.headend.module.request.entity.LogComAccess;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.ConnectorDeviceNotification;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.proto.ConnectorDeviceNotificationProto;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author liangli
 * @date 2020/9/2
 */
@Service
@Slf4j
@Transactional
public class LogTerminalAccessServiceImpl extends ServiceImpl<LogTerminalAccessMapper, LogComAccess> implements LogTerminalAccessService {
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    LogDevicePacketMapper logDevicePacketMapper;
    @Autowired
    HdComTrafficMapper hdComTrafficMapper;
    @Autowired
    LogTerminalAccessMapper logTerminalAccessMapper;
    @Autowired
    RedisUtils redisUtils;

    @Autowired
    TerminalChannelService terminalChannelService;

    @Autowired
    @Qualifier("savePacketPool")
    ThreadPoolExecutor savePool;


    @Value("${traffic.persist.on:true}")
    Boolean trafficPersist;

    @Override
    public void addAccessLog(Long termId, int type) {
        ConnectorDeviceNotification deviceNotification = new ConnectorDeviceNotification();
        deviceNotification.setComId(termId);
        deviceNotification.setTimestamp(SystemClock.now() / 1000);
        deviceNotification.setType(type == LogComAccess.AccessType.connect.getType() ?
                ConnectorDeviceNotification.NotificationType.Online.getType() : ConnectorDeviceNotification.NotificationType.Offline.getType());
        log.info("send access log: {}", deviceNotification);
        kafkaTemplate.send(TaskKafkaTopic.CON_DEVICE_NOTIFICATION, ProtoBufferUtil.toProtoBuffer(deviceNotification, ConnectorDeviceNotificationProto.ConnectorDeviceNotification.class).toByteArray());
    }

    @Override
    public void savePacket(ConnectionInfo connectionInfo, Long now, byte[] packet, Boolean toDevice) {
        savePool.submit(() -> {
            try {
                LogDevicePacket devicePacket = new LogDevicePacket();
                String packetText = ByteUtils.byteToHexString(packet);
                int direction = toDevice ? 1 : 0;
                if (connectionInfo != null) {
                    if (connectionInfo.getReturnPacket() != null && connectionInfo.getReturnPacket()) {
                        this.savePacketToRedis(connectionInfo.getTaskNo(), connectionInfo.getComId(), connectionInfo.getMeterId(), packetText, now, direction);
                    }
                    devicePacket.setComId(connectionInfo.getComId());
                    devicePacket.setMeterId(connectionInfo.getMeterId());
                    devicePacket.setTaskNo(connectionInfo.getTaskNo());
                    devicePacket.setPacketStr(packetText);
                    devicePacket.setDirection(direction);
                    devicePacket.setTv(now);
                    logDevicePacketMapper.insert(devicePacket);
                }
            } catch (Exception e) {
                log.error("savePacket failed:comId:{}, taskNo:{}", connectionInfo.getComId(), connectionInfo.getTaskNo());
            }
        });
    }

    private void savePacketToRedis(String taskNo, Long comId, Long meterId, String packet, Long now, int direction) {
        String key = RedisKeys.getTaskPacketKey(taskNo);
        TaskPacketDTO packetDTO = new TaskPacketDTO();
        packetDTO.setComId(comId);
        packetDTO.setMeterId(meterId);
        packetDTO.setHexPacket(packet);
        packetDTO.setTv(now);
        packetDTO.setDirection(direction);
        Task task = getTaskByNo(taskNo);
        if (task != null) {
            packetDTO.setMessageId(task.getMessageId());
        } else {
            log.error("get task from redis failed, taskNo={}", taskNo);
        }
        String jsonStr = JSONUtil.toJsonStr(packetDTO);
        redisUtils.lpush(key, jsonStr);
        redisUtils.expire(key, 1 * 10 * 60);
    }

    public Task getTaskByNo(String taskNo) {
        String taskContentJson = (String) redisUtils.get(RedisKeys.getTaskKey(taskNo));
        Task task = JSONUtil.toBean(taskContentJson, Task.class);
        if (task != null && StringUtils.isEmpty(task.getTaskNo())) {
            return null;
        }
        return task;
    }

    /**
     * @param channel
     * @param bytesNumber
     * @param inBoundByte 数据方向是否从系统内侧到外测
     */
    @Override
    public void saveBytesCount(Channel channel, int bytesNumber, boolean inBoundByte) {
        if (!trafficPersist) {
            return;
        }
        Attribute<Long> saveTime = channel.attr(NettyConfigHelper.saveTimestamp);
        Attribute<Long> trafficSeq = channel.attr(NettyConfigHelper.trafficSeq);
        Attribute<Integer> receiveBytes = channel.attr(NettyConfigHelper.receiveBytes);
        Attribute<Integer> sendBytes = channel.attr(NettyConfigHelper.sendBytes);
        Attribute<ComEntity> comEntityAttribute = channel.attr(NettyConfigHelper.comEntityAttrKey);

        saveTime.setIfAbsent(SystemClock.now() / 1000);
        if (inBoundByte) {
            receiveBytes.set((receiveBytes.get() == null ? 0 : receiveBytes.get()) + bytesNumber);
            sendBytes.set(sendBytes.get() == null ? 0 : sendBytes.get());
        } else {
            sendBytes.set((sendBytes.get() == null ? 0 : sendBytes.get()) + bytesNumber);
            receiveBytes.set(receiveBytes.get() == null ? 0 : receiveBytes.get());
        }

        if (null == comEntityAttribute.get() || -1 == comEntityAttribute.get().getComId()) {
            return;
        }

        if (saveTime.get() >= (SystemClock.now() / DateUtils.MILLIS_PER_SECOND) - 5 * DateUtils.MILLIS_PER_MINUTE / DateUtils.MILLIS_PER_SECOND) {
            return;
        }
        String dateStr = DateFormatUtils.format(new Date(), "yyyyMMdd");
        String comType = comEntityAttribute.get().getComType() == null ? "unknown" : comEntityAttribute.get().getComType();
        savePool.submit(() -> {
            this.saveTraffic(trafficSeq, comEntityAttribute, channel, dateStr, comType, receiveBytes, sendBytes);
        });
    }

    public void saveTraffic(Attribute<Long> trafficSeq, Attribute<ComEntity> comEntityAttribute, Channel channel, String dateStr, String comType, Attribute<Integer> receiveBytes, Attribute<Integer> sendBytes) {
        if (trafficSeq.get() == null) {
            Long comTrafficSeq = hdComTrafficMapper.getComTrafficSeq(comEntityAttribute.get().getComId(), dateStr, comType);
            if (comTrafficSeq != null) {
                channel.attr(NettyConfigHelper.trafficSeq).set(comTrafficSeq);
            }
            LoggerBuilder.getLogger(comEntityAttribute.get().getModuleSn(), this.getClass()).info("query traffic record seq {}, {}", comEntityAttribute.get().getComId(), comTrafficSeq);
        }
        HdComTraffic hdComTraffic = new HdComTraffic();
        hdComTraffic.setYearDate(dateStr);
        hdComTraffic.setSeq(channel.attr(NettyConfigHelper.trafficSeq).get());
        hdComTraffic.setInboundSum(receiveBytes.get().longValue());
        hdComTraffic.setOutboundSum(sendBytes.get().longValue());
        hdComTraffic.setPacketSize(hdComTraffic.getInboundSum() + hdComTraffic.getOutboundSum());
        hdComTraffic.setComType(comType);
        hdComTraffic.setComId(comEntityAttribute.get().getComId());
        if (hdComTraffic.getSeq() == null) {
            int seq = hdComTrafficMapper.insert(hdComTraffic);
            if (seq == 1) {
                channel.attr(NettyConfigHelper.trafficSeq).set(hdComTraffic.getSeq());
            }
        } else {
            hdComTrafficMapper.updateComTraffic(hdComTraffic);
        }
        channel.attr(NettyConfigHelper.saveTimestamp).set(SystemClock.now() / DateUtils.MILLIS_PER_SECOND);
        channel.attr(NettyConfigHelper.receiveBytes).set(0);
        channel.attr(NettyConfigHelper.sendBytes).set(0);
        LoggerBuilder.getLogger(comEntityAttribute.get().getModuleSn(), this.getClass()).info("save bytes count: {}", JSONUtil.toJsonStr(hdComTraffic));
    }
}
